package cn.gwm.flink.streaming.sink.kafka;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * @ClassName KafkaConsumerFactory
 * @Description
 * @Author LiangGuang
 * @Date 2023/06/14 14:31
 */
public class KafkaConsumerFactory {

    public static FlinkKafkaConsumer<String> getKafkaConsumer(KafkaConsumerEnum consumerEnum, String groupId, String topic) {
        FlinkKafkaConsumer<String> kafkaConsumer;
        if (KafkaConsumerEnum.EARLIEST.equals(consumerEnum)) {
            kafkaConsumer = FlinkKafkaUtil.getConsumerEarLiest(groupId, topic);
        } else if (KafkaConsumerEnum.GROUPOFFSETS.equals(consumerEnum)) {
            kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(groupId, topic);
        } else if (KafkaConsumerEnum.LATEST.equals(consumerEnum)) {
            kafkaConsumer = FlinkKafkaUtil.getConsumerLatest(groupId, topic);
        } else {
            // 指定时间戳的不提供，自己修改
            kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(groupId, topic);
        }
        return kafkaConsumer;
    }


}
